/**
 * Copyright (C) 2017-2018 pinganfang, Inc. All Rights Reserved.
 */
package com.pingan.haofang.searchcloud.register.metadata;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.ZKPaths;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.cache.Cache;
import com.pingan.haofang.searchcloud.api.matedata.IndexMeta;
import com.pingan.haofang.searchcloud.api.matedata.MetaCacheSupport;

/**
 * @author huhailiang937
 * @date 2017年7月6日下午1:31:04
 * @since 1.0.0
 */
public class MetadataChangeListener {

    protected static Logger LOG = LoggerFactory.getLogger(MetadataChangeListener.class);

    private String metaZkHost;

    private CuratorFramework client = null;

    private PathChildrenCache pathChildrenCache = null;

    public MetadataChangeListener(String metaZkHost) {
        this.metaZkHost = metaZkHost;
    }

    public void start(final Cache<String, IndexMeta> cache) {
        LOG.info("-------------> MetadataChangeListener start");
        client = CuratorFrameworkFactory.newClient(metaZkHost, new ExponentialBackoffRetry(10000, 3));
        client.start();

        pathChildrenCache = new PathChildrenCache(client, MetaCacheSupport.ROOT_META_PATH, true);
        try {
            pathChildrenCache.start();

            PathChildrenCacheListener listener = new PathChildrenCacheListener() {
                @Override
                public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                    if (null == event.getData()) {
                        return;
                    }
                    LOG.debug("--->Meta listener: " + ZKPaths.getNodeFromPath(event.getData().getPath()));
                    switch (event.getType()) {
                    case CHILD_ADDED: {
                        LOG.debug("--->Meta added: " + ZKPaths.getNodeFromPath(event.getData().getPath()));
                        if (null != event.getData().getData()) {
                            LOG.debug("--->Meta added: " + new String(event.getData().getData()));
                            addOrUpdateMeta(event.getData().getData(), cache);
                        }
                        break;
                    }

                    case CHILD_UPDATED: {
                        LOG.debug("--->Meta changed: " + ZKPaths.getNodeFromPath(event.getData().getPath()));
                        if (null != event.getData().getData()) {
                            LOG.debug("--->Meta changed: " + new String(event.getData().getData()));
                            addOrUpdateMeta(event.getData().getData(), cache);
                        }
                        break;
                    }

                    case CHILD_REMOVED: {
                        LOG.debug("--->Meta removed: " + ZKPaths.getNodeFromPath(event.getData().getPath()));
                        if (null != event.getData().getData()) {
                            LOG.debug("--->Meta removed: " + new String(event.getData().getData()));
                            removeMeta(event.getData().getData(), cache);
                        }
                        break;
                    }
                    default:
                        break;
                    }
                }
            };
            pathChildrenCache.getListenable().addListener(listener);
        } catch (Exception e) {
            LOG.error("PathChildrenCache start has error:", e);
        }
    }

    private void addOrUpdateMeta(byte[] datas, Cache<String, IndexMeta> cache) {
        try {
            IndexMeta indexMeta = MetaCacheSupport.decode(datas, IndexMeta.class);
            cache.put(indexMeta.getIndexName(), indexMeta);
        } catch (Exception e) {
            LOG.error("addOrUpdateCollectionRoute has error:", e);
        }

    }

    private void removeMeta(byte[] datas, Cache<String, IndexMeta> cache) {
        try {
            IndexMeta indexMeta = MetaCacheSupport.decode(datas, IndexMeta.class);
            cache.invalidate(indexMeta.getIndexName());
        } catch (Exception e) {
            LOG.error("removeCollectionRoute has error:", e);
        }
    }

}
